<?php

declare(strict_types=1);

namespace App\Robot\Amqp\Consumer;

use App\CustHub\Constants\SyncTaskStatusConstants;
use App\CustHub\Mapper\ChSyncTaskMapper;
use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Message\ConsumerMessage;
use Hyperf\Amqp\Result;
use Hyperf\Di\Annotation\Inject;
use PhpAmqpLib\Message\AMQPMessage;

#[Consumer(exchange:"sync_task", routingKey:"sync_task", queue:"sync_task", name:"syncTask", nums:1)]
class SyncTaskConsumer extends ConsumerMessage
{
    #[Inject]
    protected ChSyncTaskMapper $syncTaskMapper;

    public function consumeMessage($data, AMQPMessage $message): Result
    {
        $taskId = $data['taskId'];
        $taskInfo = $this->syncTaskMapper->findById($taskId);
        if(empty($taskInfo)){
            return Result::ACK;
        }

        $class = $data['className'];
        $method = $data['method'];

        if ($class && class_exists($class) && method_exists($class, $method)) {
            try {
                $this->syncTaskMapper->update($taskId,[
                    'status' => SyncTaskStatusConstants::RUNNING,
                    'start_time' => time()
                ]);
                $instance = make($class);
                $res = $instance->{$method}($taskInfo,$data['currentSeq']);
            } catch (\Throwable $throwable) {
                logger()->info('同步失败'.$throwable->getMessage());
                $this->syncTaskMapper->update($taskId,[
                    'status' => SyncTaskStatusConstants::FAILURE,
                    'failure_reason' => $throwable->getMessage(),
                    'end_time' => time()
                ]);
            }
        }
        return Result::ACK;
    }


    public function isEnable(): bool
    {
        return \Hyperf\Support\env('AMQP_CONSUMER_ENABLE',false);
    }
}